package com.jobs.producer0.controller;

import com.jobs.common.factory.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;


public class Work03 {
    private static final String TASK_QUEUE_NAME = "ack_queue";

//    public static void main(String[] args) throws Exception {
//        Channel channel = RabbitMqUtils.getChannel();
//        System.out.println("C1 等待接收消息处理时间较 短");
//
//        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
//            String message = new String(delivery.getBody());
//            try {
//                Thread.sleep(1000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            System.out.println("接收到消息:" + message);
//            /**
//             * 1. 消息标记 tag
//             * 2. 是否批量应答未应答消息
//             */
//            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
//        };
//
//        CancelCallback cancelCallback = (s) -> {
//            System.out.println(s + "消费者取消消费接口回调逻辑");
//        };
//
//        //采用手动应答
//        boolean autoAck = false;
//        channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);
//    }

    public static void main(String[] args) throws Exception {
        Work03.publishMessageAsync();
    }


    /**
     * 单个发送
     */
    public static void publishMessageIndividually() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        int messageCount = 100;

        long begin = System.currentTimeMillis();

        for (int i = 0; i < messageCount; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            //服务端返回 false 或超时时间内未返回，生产者可以消息重发
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功");
            }
        }

        long end = System.currentTimeMillis();
        System.out.println("发布" + messageCount + "个单独确认消息,耗时" + (end - begin) + "ms");
    }

    /**
     * 批量
     */
    public static void publishMessageBatch() throws Exception {
        int MESSAGE_COUNT = 200;
        Channel channel = RabbitMqUtils.getChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        //批量确认消息大小
        int batchSize = 100;
        //未确认消息个数
        int outstandingMessageCount = 0;
        long begin = System.currentTimeMillis();

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            outstandingMessageCount++;
            if (outstandingMessageCount == batchSize) {
                channel.waitForConfirms();
                outstandingMessageCount = 0;
            }
        }
        //为了确保还有剩余没有确认消息 再次确认
        if (outstandingMessageCount > 0) {
            channel.waitForConfirms();
        }
        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + (end - begin) + "ms");
    }


    //    异步确认发布
    private static void publishMessageAsync() throws Exception {

        int MESSAGE_COUNT = 200;
        Channel channel = RabbitMqUtils.getChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();

        /**线程安全有序的一个哈希表，适用于高并发的情况下
         1.轻松的将序号与消息进行关联
         2.轻松批量删除条目，只要给到序号
         3.支持高并发（多线程）
         **/
        //消息确认成功，回调函数
        ConcurrentSkipListMap<Long,String> outstandingConfirms =
                new ConcurrentSkipListMap<>();
        ConfirmCallback ackCallback = (deliveryTag , multiple)->{
            //确认了多少条，multiple:批量或单个
            if (multiple){
                //批量的
                //2.轻松批量删除条目，只要给到序号
                ConcurrentNavigableMap<Long,String> confirmed =
                        outstandingConfirms.headMap(deliveryTag);
                //消息的序号
                confirmed.clear();
            }else {  //单个的
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("确认的消息："+deliveryTag);
        };

        //消息确认失败，回调函数
        /**
         * 1.消息的标记
         * 2.是否为批量确认
         */
        ConfirmCallback nackCallback = (deliveryTag , multiple)->{
            //3、打印一下未确认的消息都有哪些
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息是:"+message+":::未确认的消息："+deliveryTag);
        };


        //准备消息的监听器 ，监听哪些消息成功了， 哪些失败了
        channel.addConfirmListener( ackCallback, nackCallback);

        //开始时间
        long begin = System.currentTimeMillis();

        //发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            //1.轻松的将序号与消息进行关联，将每一条消息都存放hash表里面
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
        }

        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息,耗时" + (end - begin) + "ms");
    }



}

